HDFS JAVA API

Hadoop 中的文件操作类基本上是在“org.apache.hadoop.fs”包中，这些 api 能够支持操作包含了创建文件、读写文件和删除文件等。

API 的使用

FileSystem 类

FileSystem 是所有 HDFS 操作的主入口，封装了几乎所有的文件操作，比如 mkdir、delete等。由于之后的每个单元测试都需要用到它，这里使用 @Before 注解进行标注。

private static final String HDFS_PATH = "hdfs://192.168.3.11:9000";
private static final String HDFS_USER = "root";
private static FileSystem fileSystem;

@Before
public void prepare() {
    try {
        Configuration configuration = new Configuration();
        // 这里我启动的是单节点的 Hadoop,所以副本系数设置为 1,默认值为 3
        configuration.set("dfs.replication", "1");
        fileSystem = FileSystem.get(new URI(HDFS_PATH), configuration, HDFS_USER);
    } catch (IOException e) {
        e.printStackTrace();
    } catch (InterruptedException e) {
        e.printStackTrace();
    } catch (URISyntaxException e) {
        e.printStackTrace();
    }
}


@After
public void destroy() {
    fileSystem = null;
}

创建目录，支持递归创建

@Test
public void mkDir() throws Exception {
    boolean mkdirs = fileSystem.mkdirs(new Path("/hdfs-api/test0/"));
    System.out.println(mkdirs);
}

创建具有指定权限的目录

@Test
public void mkDirWithPermission() throws Exception {
    boolean mkdirs = fileSystem.mkdirs(new Path("/hdfs-api/test1/"),
            new FsPermission(FsAction.READ_WRITE, FsAction.READ, FsAction.READ));
    System.out.println(mkdirs);
}

创建文件，并写入内容

@Test
public void create() throws Exception {
    // 如果文件存在，默认会覆盖, 可以通过第二个参数进行控制。第三个参数可以控制使用缓冲区的大小
    FSDataOutputStream out = fileSystem.create(new Path("/hdfs-api/test/a.txt"),
            true, 4096);
    out.write("Hello Hadoop!".getBytes());
    out.write("Hello Spark!".getBytes());
    // 强制将缓冲区中内容刷出
    out.flush();
    out.close();
}

判断文件是否存在

@Test
public void exist() throws Exception {
    boolean exists = fileSystem.exists(new Path("/hdfs-api/test/a.txt"));
    System.out.println(exists);
}

查看文件内容

@Test
public void readToString() throws Exception {
    FSDataInputStream inputStream = fileSystem.open(new Path("/hdfs-api/test/a.txt"));
    String context = inputStreamToString(inputStream, "utf-8");
    System.out.println(context);
}

/**
* 把输入流转换为指定编码的字符
*/
private static String inputStreamToString(InputStream inputStream, String encode) {
    try {
        if (encode == null || ("".equals(encode))) {
            encode = "utf-8";
        }
        BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, encode));
        StringBuilder builder = new StringBuilder();
        String str = "";
        while ((str = reader.readLine()) != null) {
            builder.append(str).append("\n");
        }
        return builder.toString();
    } catch (IOException e) {
        e.printStackTrace();
    }
    return null;
}

文件重命名

@Test
public void rename() throws Exception {
    Path oldPath = new Path("/hdfs-api/test/a.txt");
    Path newPath = new Path("/hdfs-api/test/b.txt");
    boolean result = fileSystem.rename(oldPath, newPath);
    System.out.println(result);
}

删除文件

@Test
public void delete() throws Exception {
    /*
    *  第二个参数代表是否递归删除
    *  如果path是一个目录且递归删除为true, 则删除该目录及其中所有文件;
    *  如果path是一个目录但递归删除为false,则会则抛出异常。
    */
    boolean result = fileSystem.delete(new Path("/hdfs-api/test/b.txt"), true);
    System.out.println(result);
}

上传文件到HDFS

@Test
public void copyFromLocalFile() throws Exception {
    // 如果指定的是目录，则会把目录及其中的文件都复制到指定目录下
    Path src = new Path("/Users/zhengweilu/Downloads/typora-onedark-theme.zip");
    Path dst = new Path("/hdfs-api/test/");
    fileSystem.copyFromLocalFile(src, dst);
}

上传文件到HDFS

@Test
public void copyFromLocalBigFile() throws Exception {

    File file = new File("/Users/zhengweilu/projects/private/open-source-server/bigdata/hadoop-2.10.0.tar.gz");
    final float fileSize = file.length();
    InputStream in = new BufferedInputStream(new FileInputStream(file));

    FSDataOutputStream out = fileSystem.create(new Path("/hdfs-api/test/hadoop-2.10.0.tar.gz"),
            new Progressable() {
                long fileCount = 0;

                public void progress() {
                    fileCount++;
                    // progress方法每上传大约64KB的数据后就会被调用一次
                    System.out.println("文件上传总进度：" + (fileCount * 64 * 1024 / fileSize) * 100 + " %");
                }
            });

    IOUtils.copyBytes(in, out, 4096);
}

从HDFS上下载文件

@Test
public void copyToLocalFile() throws Exception {
    Path src = new Path("/hdfs-api/test/typora-onedark-theme.zip");
    Path dst = new Path("/Users/zhengweilu/Downloads/");
    /*
    * 第一个参数控制下载完成后是否删除源文件,默认是true,即删除;
    * 最后一个参数表示是否将RawLocalFileSystem用作本地文件系统;
    * RawLocalFileSystem默认为false,通常情况下可以不设置,
    * 但如果你在执行时候抛出NullPointerException异常,则代表你的文件系统与程序可能存在不兼容的情况(window下常见),
    * 此时可以将RawLocalFileSystem设置为true
    */
    fileSystem.copyToLocalFile(false, src, dst, true);
}

查看指定目录下所有文件的信息

@Test
public void listFiles() throws Exception {
    FileStatus[] statuses = fileSystem.listStatus(new Path("/hdfs-api"));
    for (FileStatus fileStatus : statuses) {
        //fileStatus的toString方法被重写过，直接打印可以看到所有信息
        System.out.println(fileStatus.toString());
    }
}

递归查看指定目录下所有文件的信息

@Test
public void listFilesRecursive() throws Exception {
    RemoteIterator<LocatedFileStatus> files = fileSystem.listFiles(new Path("/"), true);
    while (files.hasNext()) {
        System.out.println(files.next());
    }
}

查看文件块信息

@Test
public void getFileBlockLocations() throws Exception {

    FileStatus fileStatus = fileSystem.getFileStatus(new Path("/hdfs-api/test/hadoop-2.10.0.tar.gz"));
    BlockLocation[] blocks = fileSystem.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
    for (BlockLocation block : blocks) {
        System.out.println(Arrays.toString(block.getHosts()) + "\t" + block.getLength() + "\t" + Arrays.toString(block.getNames()));
    }
}

测试代码

TDB

这不是个BUG，这是一个未注明的功能特征。